package com.greate.community.config;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;

import java.util.HashMap;
import java.util.Map;

/**
 * @author keke  kafka消费事件配置类
 */
@Configuration
public class KafkaConsumerConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${spring.kafka.consumer.session-timeout-ms}")
    private String sessionTimeoutMs;

    @Value("${spring.kafka.consumer.max-poll-interval-ms}")
    private String maxPollIntervalMs;

    @Value("${spring.kafka.consumer.auto-offset-reset}")
    private String autoOffsetReset;

    /**
     * 动态创建 ConsumerFactory
     *
     * @param groupId 消费者组 ID
     * @return ConsumerFactory
     */

    /*
    在运行时动态创建 ConsumerFactory 实例，而不是由 Spring 容器直接管理，那么不需要添加 @Bean。
    如果需要创建一个固定的实例，并且希望由 Spring 容器管理其生命周期，那么需要添加 @Bean。
    * */
    public ConsumerFactory<String, String> createConsumerFactory(String groupId) {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); // Kafka 服务器地址
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId); // 消费者组 ID
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); // Key 反序列化器
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); // Value 反序列化器
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); // 禁用自动提交偏移量
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeoutMs); // 会话超时时间
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, maxPollIntervalMs); // 最大拉取间隔时间
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset); // 偏移量重置策略
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100); // 每次最多拉取 100 条消息（可以设成参数）
        return new DefaultKafkaConsumerFactory<>(props);
    }

    /**
     * 默认 ConsumerFactory
     */
    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        return createConsumerFactory("defaultConsumerGroup");
    }

    /**
     * 创建 KafkaConsumer 实例
     *
     * @param groupId 消费者组 ID
     * @return KafkaConsumer 实例
     */
    public KafkaConsumer<String, String> createConsumer(String groupId) {
        ConsumerFactory<String, String> consumerFactory = createConsumerFactory(groupId);
        return (KafkaConsumer<String, String>) consumerFactory.createConsumer();
    }
}

